#include <unistd.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "master.h"
#include "debug.h"


struct Master *master;


int parse_master_conf(char *conf)
{
	debug(SECTION_MASTER, 10, "---> parse_master_conf\n");
	int ret = 0;
	FILE *fp;
	char opt_name[100];
	char opt_val[100];

	if ((fp = fopen(conf, "r")) != NULL) {
		while (fscanf(fp, "%s\t=\t%s", opt_name, opt_val) != EOF) {
			debug(SECTION_MASTER, 10, "CONFIG : %s = %s\n", opt_name, opt_val);
			if (!strcmp(opt_name, "master_id2uri")) {
				//TODO when master's number isn't one
				char id[5];
				char uri[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id, uri);
				master->id = atoi(id);
				master->uri = strdup(uri);
			} else if (!strcmp(opt_name, "cs_num")) {
				master->cs_num = atoi(opt_val);
				master->chunkservers = (struct ChunkServer **)malloc(sizeof(struct ChunkServer *) * master->cs_num);
				master->heartbeatmap = (struct HeartbeatInfo **)malloc(sizeof(struct HeartbeatInfo *) * master->cs_num);
				memset(master->chunkservers, 0, sizeof(struct ChunkServer *) * master->cs_num);
				memset(master->heartbeatmap, 0, sizeof(struct HeartbeatInfo *) * master->cs_num);
			} else if (!strcmp(opt_name, "cs_id2uri")) {
				char id[5];
				char uri[30];
				sscanf(opt_val, "%[0-9]-%[0-9,:,.]", id, uri);
				uint64_t cs_id = atoi(id);
				if (cs_id >= master->cs_num) {
					ret = -1;
					return ret;
				}
				struct ChunkServer *cs = (struct ChunkServer *)malloc(sizeof(struct ChunkServer));
				memset(cs, 0, sizeof(struct ChunkServer));
				cs->id = cs_id;
				cs->uri = strdup(uri);
				cs->state = DOWN;
				//FIXME : check if cs is already exist
				master->chunkservers[cs->id] = cs;
				struct HeartbeatInfo *hinfo = (struct HeartbeatInfo *)malloc(sizeof(struct HeartbeatInfo));
				memset(hinfo, 0, sizeof(struct HeartbeatInfo));
				hinfo->chunkserver = cs->id;
				master->heartbeatmap[cs->id] = hinfo;
			} else if (!strcmp(opt_name, "heartbeat_interval")) {
				master->heartbeat_interval = atoi(opt_val);
			} else {
				debug(SECTION_MASTER, 10, "Unrecognized config option : %s\n", opt_name);
			}
		}
		fclose(fp);
	} else {
		ret = -1;
		debug(SECTION_MASTER, 10, "Failed to open conf file : %s", conf);
		return ret;
	}
	return ret;
}

/**
 * arg1: struct rpc * (master->rpc_server)
 * arg2: struct rpc_execution *
 * arg3: struct dss_msg *
**/
void m_chunk_op_service(void *arg1, void *arg2, void *arg3)
{
	debug(SECTION_MASTER, 10, "---> m_chunk_op_service\n");
	struct rpc_execution *ex = (struct rpc_execution *)arg2;
	struct msg_base* rpc_req = get_request_header(ex);
	struct msg_base* rpc_resp = get_response_header(ex);
	struct dss_msg *msg = (struct dss_msg *)(rpc_req->data);

	//TODO alloc mem from rpc_client??
	struct ChunkOp* op = alloc_small_piece(master->rpc_client);
	memset(op, 0, sizeof(struct ChunkOp));
	op->rpc_ex = ex;
	op->dss_request = msg;
	op->dss_response = (struct dss_msg*)(rpc_resp->data);

	switch (msg->msg_type)
	{
	case CHUNK_OP_CREATE:
		m_handle_op_create(op);
		break;
	case CHUNK_OP_LIST:
		m_handle_op_list(op);
		break;
	default:
		//TODO unexpect message
		debug(SECTION_MASTER, 10, "unexpect message\n");
	}
	debug(SECTION_MASTER, 10, "<--- m_chunk_op_service\n");
	return;
}
/**
 * arg1: struct rpc * (master->rpc_server)
 * arg2: struct rpc_execution *
 * arg3: struct dss_msg *
**/
void m_sys_status_service(void *arg1, void *arg2, void *arg3)
{
	debug(SECTION_MASTER, 10, "---> m_sys_status_service\n");
	debug(SECTION_MASTER, 10, "<--- m_sys_status_service\n");
}

/**
 * arg1: struct rpc * (master->rpc_server)
 * arg2: struct rpc_execution *
 * arg3: struct dss_msg *
**/
void m_heartbeat_service(void *arg1, void *arg2, void *arg3)
{
	debug(SECTION_MASTER, 10, "---> m_heartbeat_service\n");
	struct rpc_execution *ex = (struct rpc_execution *)arg2;
	struct msg_base *rpc_req = get_request_header(ex);
	struct msg_base *rpc_resp = get_response_header(ex);
	struct dss_msg *msg = (struct dss_msg *)rpc_req->data;
	struct dss_msg *heart_resp = (struct dss_msg *)rpc_resp->data;
	struct heartbeat_req *heart_header = (struct heartbeat_req *)msg->data;

	//1. update master->chunkservers; 2. update master->heartbeatmap
	struct ChunkServer *cs = master->chunkservers[heart_header->cs_id];
	struct HeartbeatInfo *heart_info = master->heartbeatmap[heart_header->cs_id];

	heart_info->last_heartbeat = rdtsc();
	heart_info->last_tid = msg->msg_id;
	cs->cluster_total = heart_header->cluster_total;
	cs->cluster_free = heart_header->cluster_free;
	cs->cluster_size = heart_header->cluster_size;
	cs->io_unit_size = heart_header->io_unit_size;

	//set value to heartbeat_response
	heart_resp->status_code = SUCCESS;
	heart_resp->msg_type = CS_HEARTBEAT_REPLY;
	heart_resp->msg_id = msg->msg_id;
	heart_resp->data_len = 0;
	rpc_resp->data_len = sizeof(struct dss_msg) + heart_resp->data_len;
	rpc_handle_complete(master->rpc_server, ex);
	debug(SECTION_MASTER, 10, "<--- m_heartbeat_service\n");

}

void m_handle_op_create(struct ChunkOp *op)
{
	debug(SECTION_MASTER, 10, "---> m_handle_op_create\n");
	op->total_response = 1;
	op->cur_response = 0;
	//to select a chunkserver to create chunk
	int sess_id = chunkserver_select();
	if (sess_id < 0) {
		//set value to create_response
		op->dss_response->status_code = FAILURE;
		op->dss_response->msg_type = CHUNK_OP_CREATE_REPLY;
		op->dss_response->msg_id = op->dss_request->msg_id;
		op->dss_response->data_len = 0;
		get_response_header(op->rpc_ex)->data_len = sizeof(struct dss_msg) + op->dss_response->data_len;
		rpc_handle_complete(master->rpc_server, op->rpc_ex);
		free_small_piece(master->rpc_client, op);
		debug(SECTION_MASTER, 10, "no accessible chunkserver!!!!!!!!!\n");
		return;
	}
	//construct a new request to chunkserver to create chunk
	struct msg_base* rpc_header = alloc_msg_header(master->rpc_client);

	prepare_msg_header(master->rpc_client, rpc_header, S_CHUNK_IO_SERVICE, 0, 0, 0, 0);

	struct dss_msg* dss_header = (struct dss_msg*)rpc_header->data;
	dss_header->client_id = -1;
	dss_header->msg_type = CHUNK_OP_CREATE;
	dss_header->msg_id = master->cur_id++;
	dss_header->data_len = 0;

	rpc_header->data_len = sizeof(struct dss_msg) + dss_header->data_len;
	//master_create_cb, op
	enqueue_request(master->rpc_client, sess_id, rpc_header, (rpc_request_func)master_create_cb, op, 0);
	debug(SECTION_MASTER, 10, "<--- m_handle_op_create\n");
}

void m_handle_op_list(struct ChunkOp * op)
{
	debug(SECTION_MASTER, 10, "---> m_handle_op_list\n");
	int i = 0;
	op->dss_response->msg_type = CHUNK_OP_LIST_REPLY;
	op->total_response = master->cs_num;
	op->cur_response = 0;
	
	struct msg_base *rpc_header = alloc_msg_header(master->rpc_client);

	prepare_msg_header(master->rpc_client, rpc_header, S_CHUNK_IO_SERVICE, 0, 0, 0, 0);

	struct dss_msg* dss_header = (struct dss_msg*)rpc_header->data;
	dss_header->client_id = -1;
	dss_header->msg_id = master->cur_id++;
	dss_header->msg_type = CHUNK_OP_LIST;
	dss_header->data_len = 0;

	rpc_header->data_len = sizeof(struct dss_msg) + dss_header->data_len;
	//to each chunkserver construct a new request to get chunk list
	for (i = 0; i < master->cs_num; i++) {
		//TODO master_list_cb
		enqueue_request(master->rpc_client, master->chunkservers[i]->session_id, rpc_header, (rpc_request_func)master_list_cb, op, 0);
	}
	debug(SECTION_MASTER, 10, "<--- m_handle_op_list\n");
}

int chunkserver_select()
{
	debug(SECTION_MASTER, 10, "---> chunkserver_select\n");
	int cs_id = 0;
	int count = 0;
	do {
		cs_id = master->cur_cs % master->cs_num;
		master->cur_cs++;
		if (master->cur_cs >= master->cs_num)
			master->cur_cs = 0;
		count++;
		if (count > master->cs_num)
			return -1;
	} while (master->chunkservers[cs_id]->state == DOWN);
	
	struct ChunkServer *cs = master->chunkservers[cs_id];
	debug(SECTION_MASTER, 10, "<--- chunkserver_select, cs_id = %d, session_id = %d\n", cs_id, cs->session_id);
	return cs->session_id;
}

//struct rpc* self, struct rpc_execution* ex, void* usr_ctx
void master_create_cb(void *arg1, void *arg2, void *ctx)
{
	debug(SECTION_MASTER, 10, "---> master_create_cb\n");
	struct rpc_execution *ex = (struct rpc_execution *)arg2;
	struct msg_base *rpc_resp = get_response_header(ex);
	struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;
	struct ChunkOp *op = (struct ChunkOp *)ctx;
	op->cur_response++;
	if (op->cur_response == op->total_response) {
		memcpy(op->dss_response, dss_resp, rpc_resp->data_len);
		struct chunk_resp *create_resp = (struct chunk_resp *)op->dss_response->data;
		debug(SECTION_MASTER, 10, "chunkid :[%d, %d]\n", create_resp->id.cs_id, create_resp->id.blob_id);
		get_response_header(op->rpc_ex)->data_len = sizeof(struct dss_msg) + sizeof(struct chunk_resp);
		//op->rpc_ex->slot.response_header->data_len = sizeof(struct dss_msg) + sizeof(struct chunk_resp);
		free_small_piece(master->rpc_client, get_request_header(ex));
		rpc_handle_complete(master->rpc_server, op->rpc_ex);
		free_small_piece(master->rpc_client, op);
	}
	debug(SECTION_MASTER, 10, "<--- master_create_cb\n");
}

void master_list_cb(void *arg1, void *arg2, void *ctx)
{
	debug(SECTION_MASTER, 10, "---> master_list_cb\n");
	struct rpc_execution *ex = (struct rpc_execution *)arg2;
	struct msg_base *rpc_resp = get_response_header(ex);
	struct dss_msg *dss_resp = (struct dss_msg *)rpc_resp->data;
	struct ChunkOp *op = (struct ChunkOp *)ctx;
	op->cur_response++;
	if (op->cur_response == op->total_response) {
	//TODO 
	}
	debug(SECTION_MASTER, 10, "<--- master_list_cb\n");
}


void master_prepare() {
	debug(SECTION_MASTER, 10, "---> master_prepare\n");
	int i = 0;
	//create rpc
	master->rpc_client = rpc_create(NULL, 0, RPC_RDMA | RPC_SHARED_RECV_QUEUE, NULL);
	master->rpc_server = rpc_create(NULL, 0, RPC_SERVER | RPC_RDMA | RPC_SHARED_RECV_QUEUE, NULL);
	rpc_bind_local_uri(master->rpc_server, master->uri);
	/* create rpc session from master to each chunkserver
	for (i = 0; i < master->cs_num; i++) {
		master->chunkservers[i]->session_id = rpc_create_session(master->rpc_client, master->chunkservers[i]->uri, NULL, NULL);
		debug(SECTION_MASTER, 10, "cs: %d, session_id: %d\n", i, master->chunkservers[i]->session_id);
	}
	*/
	rpc_register_on_request(master->rpc_server, M_CHUNKOP_SERVICE, (rpc_request_func)m_chunk_op_service);
	rpc_register_on_request(master->rpc_server, M_SYSSTATUS_SERVICE, (rpc_request_func)m_sys_status_service);
	rpc_register_on_request(master->rpc_server, M_HEARTBEAT_SERVICE, (rpc_request_func)m_heartbeat_service);
	master->state = ACTIVE;

	debug(SECTION_MASTER, 10, "<--- master_prepare\n");
}

void master_exit() {
	debug(SECTION_MASTER, 10, "---> master_exit\n");
	//TODO close rpc, free memory... 
	debug(SECTION_MASTER, 10, "<--- master_exit\n");

}

void * heartbeat_main(void *arg)
{
	debug(SECTION_MASTER, 10, "---> heartbeat_main\n");
	int i = 0;
	void *ret = NULL;
	struct ChunkServer *cs = NULL;
	while (master->state != EXIT) {
		if (master->state == ACTIVE) {
			// check every chunkserver's heartbeat and update it's state
			for (i = 0; i < master->cs_num; i++) {
				debug(SECTION_MASTER, 10, "cs:%d, last_heartbeat:%d\n", i, master->heartbeatmap[i]->last_heartbeat);
				uint64_t now = rdtsc();
				uint64_t last_interval = to_seconds(now - master->heartbeatmap[i]->last_heartbeat, 0);
				debug(SECTION_MASTER, 10, "now : %d\n", now);
				debug(SECTION_MASTER, 10, "last_interval : %d\n", last_interval);
				if(to_seconds(rdtsc() - master->heartbeatmap[i]->last_heartbeat, 0) > master->heartbeat_interval) {
					master->chunkservers[i]->state = DOWN;
				} else if (master->chunkservers[i]->state == DOWN) {
					debug(SECTION_MASTER, 10, "chunkserver: %d up\n", i);
					//TODO : check rpc_create_session's return value!!!
					master->chunkservers[i]->session_id = rpc_create_session(master->rpc_client, master->chunkservers[i]->uri, NULL, NULL);
					master->chunkservers[i]->state = UP;
				}
				debug(SECTION_MASTER, 10, "cs:%d, state:%s\n", i, master->chunkservers[i]->state ? "UP" : "DOWN");
			}
		} else {
			//nothing to do?
			debug(SECTION_MASTER, 10, "master's state is not active!!!\n");
		}
		debug(SECTION_MASTER, 10, "heartbeat_interval = %d\n", master->heartbeat_interval);
		sleep(master->heartbeat_interval);
	}
	debug(SECTION_MASTER, 10, "<--- heartbeat_main\n");
	return ret;
}

int main(int argc, char **argv)
{
	init_debug_levels(SECTION_MASTER, 20);
	master = (struct Master *)malloc(sizeof(struct Master));
	memset(master, 0, sizeof(struct Master));

	int ch;
	int ret = 0;
	while ((ch = getopt(argc, argv, "i:c:h")) != -1) {
		switch (ch) {
		case 'i':
			master->id = atoi(optarg);
			break;
		case 'c':
			master->conf = strdup(optarg);
			break;
		case 'h':
			debug(SECTION_MASTER, 10, "\n-i Master ID\n-c DSS Configure File Name\n-h Help\n");
			return ret;
		default:
			debug(SECTION_MASTER, 10, "NOT recognized option: %c", ch);
			debug(SECTION_MASTER, 10, "\n-i Master ID\n-c DSS Configure File Name\n-h Help\n");
			return ret;
		}
	}
	//init master from configure file
	if (0 != parse_master_conf(master->conf)) {
		debug(SECTION_MASTER, 10, "parse configure file err\n");
		return ret;
	}
	
	// start heartbeat thread
	pthread_create(&(master->heartbeat_th), 0, heartbeat_main, NULL);

	//rpc ... prepare
	master_prepare();

	//accept rpc connection	and request
	uint64_t now = 0;
	while(master->state == ACTIVE)
	{
		now = rdtsc();
		if (to_seconds(now - master->last_active, 0) > 1) {
			rpc_accept_session(master->rpc_server);
			master->last_active = now;
		}
		run_poll_once(master->rpc_server);
		run_poll_once(master->rpc_client);
	}

	//TODO exit
	pthread_join(master->heartbeat_th, NULL); //need move in master_exit()
	master_exit();

	return 0;
}



